package com.lee.mq.mqserver;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: lishuo
 * Date: 2023-10-26
 * Time: 10:39
 */

import com.lee.mq.common.Consumer;
import com.lee.mq.common.MqException;
import com.lee.mq.mqserver.core.*;
import com.lee.mq.mqserver.datacenter.DiskDataCenter;
import com.lee.mq.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 这个类表示虚拟主机，用来统一管理内存和硬盘上的数据，并对外提供核心业务API
 */
public class VirtualHost {
    // 虚拟主机名
    private String virtualHostName;
    // 创建操作硬盘实例
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    // 创建操作内存实例
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    // 规则验证
    private Router router = new Router();

    // 交换机锁对象
    private final Object exchangeLocker = new Object();
    // 对列锁对象
    private final Object queueLocker = new Object();

    // 创建消费处理实例
    private ConsumerManager consumerManager = new ConsumerManager(this);



    // 对外提供get方法
    public String getVirtualHostName() {
        return virtualHostName;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    // 构造方法
    public VirtualHost(String name) {
        this.virtualHostName = name;
        // MemoryDataCenter 不需要进行额外初始化操作
        // DiskDataCenter 需要进行建库建表数据初始化操作
        diskDataCenter.init();

        // 恢复硬盘数据到内存
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败!");
        }
    }

    /**
     * 创建交换机
     * @param exchangeName
     * @param exchangeType
     * @param durable
     * @param autoDelete
     * @param arguments
     * @return true 创建成功，false 创建失败
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType,boolean durable,boolean autoDelete,
                                   Map<String,Object> arguments) {
        try {
            synchronized (exchangeLocker) {
                exchangeName = virtualHostName + exchangeName;
                // 1.判断交换机是否已存在
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    System.out.println("[VirtualHost] 交换机已存在: exchangeName = "+exchangeName);
                    return true;
                }
                // 2.封装交换机对象
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                // 3.持久化：写入数据库
                if (durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                // 4.写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建成功: exchangeName = "+exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机创建失败: exchangeName = "+exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 删除交换机
     * @param exchangeName 交换机名
     * @return true 删除成功，false 删除失败
     */
    public boolean exchangeDelete(String exchangeName) {
        try {
            synchronized (exchangeLocker) {
                exchangeName = virtualHostName + exchangeName;
                // 1.判断交换机是否存在
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange == null) {
                    throw new MqException("[VirtualHost] 交换机不存在无法删除 exchangeName = "+exchangeName);
                }
                // 2.删除硬盘上的数据
                if (existsExchange.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                // 3.删除内存数据
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功: exchangeName = "+exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败: exchangeName = "+exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 创建队列
     * @param queueName
     * @param durable
     * @param exclusive
     * @param autoDelete
     * @param arguments
     * @return true 创建成功，false 创建失败
     */
    public boolean queueDeclare(String queueName,boolean durable,boolean exclusive,boolean autoDelete,
                                Map<String,Object> arguments) {
        try {
            synchronized (queueLocker) {
                queueName = virtualHostName + queueName;
                // 1.判断队列是否存在
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已存在: queueName = "+queueName);
                    return true;
                }
                // 2.封装队列对象
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                // 3.持久化: 写入数据库
                diskDataCenter.insetQueue(queue);
                // 4.写入内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功: queueName = "+queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败: queueName = "+queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 删除对列
     * @param queueName 队列名
     * @return true 删除成功，false 删除失败
     */
    public boolean queueDelete(String queueName){
        try {
            synchronized (queueLocker) {
                queueName = virtualHostName + queueName;
                // 1.判断对列是否存在
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue == null) {
                    throw new MqException("[VirtualHost] 队列不存在无法删除 queueName = "+queueName);
                }
                // 2.删除硬盘数据
                if (existsQueue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                // 3.删除内存数据
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 队列删除成功: queueName = "+queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列删除失败: queueName = "+queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 创建绑定
     * @param exchangeName
     * @param queueName
     * @param bindingKey
     * @return true 创建成功，false 创建失败
     */
    public boolean queueBind(String exchangeName,String queueName,String bindingKey) {
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    exchangeName = virtualHostName + exchangeName;
                    queueName = virtualHostName + queueName;
                    // 1.判断绑定是否存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding != null) {
                        System.out.println("[VirtualHost] 绑定已存在无法创建 exchangeName = "+exchangeName+", queueName = "+queueName);
                        return true;
                    }
                    // 2.判断bindingKey是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] bindingKey 非法：bindingKey = "+bindingKey);
                    }
                    // 3.判断交换机是否存在
                    Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                    if (existsExchange == null) {
                        throw new MqException("[VirtualHost] 创建绑定时，交换机不存在: exchangeName = "+exchangeName);
                    }
                    // 4.判断队列是否存在
                    MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                    if (existsQueue == null) {
                        throw new MqException("[VirtualHost] 创建绑定时，队列不存在: queueName = "+queueName);
                    }
                    // 5.封装绑定对象
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    // 6.持久化：写入数据库
                    if(existsExchange.isDurable() && existsQueue.isDurable()) {
                        diskDataCenter.insetBinding(binding);
                    }
                    // 7.写入内存
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 绑定创建成功: exchangeName = "+exchangeName+", queueName = "+queueName);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定创建失败: exchangeName = "+exchangeName+", queueName = "+queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 解除绑定
     * @param exchangeName
     * @param queueName
     * @return true 删除成功，false 删除失败
     */
    public boolean queueUnbind(String exchangeName,String queueName) {
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    exchangeName = virtualHostName + exchangeName;
                    queueName = virtualHostName + queueName;
                    // 1.判断绑定是否存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding == null) {
                        throw new MqException("[VirtualHost] 绑定不存在无法删除  exchangeName = "+exchangeName+", queueName = "+queueName);
                    }
                    // 2.删除硬盘数据（不管是否持久化都没关系）
                    diskDataCenter.deleteBinding(existsBinding);
                    // 3.删除内存数据
                    memoryDataCenter.deleteBinding(existsBinding);
                    System.out.println("[VirtualHost] 绑定删除成功: exchangeName = "+exchangeName+", queueName = "+queueName);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定删除失败: exchangeName = "+exchangeName+", queueName = "+queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 通过交换机发送消息到指定队列
     * @param routingKey
     * @param body
     * @return
     */
    public boolean basicPublish(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body) {
        try {
            exchangeName = virtualHostName + exchangeName;
            // 1.检测routingKey是否合法
            if (!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey 非法: routingKey = "+routingKey);
            }
            // 2.判断交换机是否存在
            Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
            if (existsExchange == null) {
                throw new MqException("[VirtualHost] 交换机不存在: exchangeName = "+exchangeName);
            }
            // 3.判断交换机类型
            if (existsExchange.getType() == ExchangeType.DIRECT) {
                // 直接交换机单独处理.
                // （1）将routingKey当做队列名(无视绑定关系)，查询队列
                String queueName = virtualHostName + routingKey;
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue == null) {
                    throw new MqException("[VirtualHost] 对列不存在: queueName = "+existsQueue.getName());
                }
                // （2）封装消息
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                // （3）发送消息
                sendMessage(existsQueue,message);

            } else {
                // 扇出交换机和主题交换机统一处理（处理方式类似）
                // （1）根据交换机查询绑定、
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                if (bindingsMap == null) {
                    throw new MqException("[VirtualHost] 队列不存在任何绑定: exchangeName = "+ exchangeName);
                }
                // （2）遍历绑定发送消息
                for (Map.Entry<String,Binding> bindingMap : bindingsMap.entrySet()) {
                    // 获取队列
                    Binding binding = bindingMap.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        // 此处队列为空，跳过本次发送
                        System.out.println("[VirtualHost] 对列不存在: queueName = "+queue.getName());
                        continue;
                    }
                    // 封装消息
                    Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                    // 3) 判定这个消息是否能转发给该队列.
                    //    如果是 fanout, 所有绑定的队列都要转发的.
                    //    如果是 topic, 还需要判定下, bindingKey 和 routingKey 是不是匹配.
                    if (!router.route(existsExchange.getType(),binding,message)) {
                        continue;
                    }
                    // 发送消息
                    sendMessage(queue,message);
                }
            }
            System.out.println("[VirtualHost] 发送消息成功");
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 发送消息失败");
            e.printStackTrace();
            return false;
        }
    }

    // 封装发送消息的方法
    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        // 持久化: 写入硬盘
        if (message.getDeliverMode() == 2) {
            diskDataCenter.sendMessage(queue,message);
        }
        // 写入内存
        memoryDataCenter.sendMessage(queue.getName(),message);

        // 通知消费者（订阅者）可以消费消息了
        consumerManager.notifyConsume(queue.getName());
    }

    /**
     * 订阅消息（在指定的队列中添加订阅者）
     * @param consumerTag 消费者的唯一身份标识
     * @param queueName 订阅的队列名
     * @param autoAsk 应答方式：true 自动应答，false 手动应答
     * @param consumer 函数式接口：回调函数，表示具体的消息处理业务逻辑
     * @return true 订阅成功 ， false 订阅失败
     */
    public boolean basicCosume(String consumerTag, String queueName, boolean autoAsk, Consumer consumer) {
        try {
            // 调用addConsumerEnv
            consumerManager.addConsumerEnv(consumerTag,queueName,autoAsk,consumer);
            System.out.println("[VirtualHost] basicConsume 成功! queueName=" + queueName);
            return true;
        } catch (MqException e) {
            System.out.println("[VirtualHost] basicConsume 失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 应答消息（手动应答）
     * @param queueName
     * @param messageId
     * @return
     */
    public boolean basicAck(String queueName,String messageId) {
        queueName = virtualHostName + queueName;
        try {
            // 1.查询队列，消息是否存在
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MqException("[VirtualHost] 队列不存在 queueName = "+queueName);
            }
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null) {
                throw new MqException("[VirtualHost] 消息中心的消息不存在 queueName = "+queueName);
            }
            message = memoryDataCenter.getMessageWaitAsk(queueName,messageId);
            if (message == null) {
                throw new MqException("[VirtualHost] 消息待应答的消息不存在 queueName = "+queueName);
            }
            // 2.删除硬盘上的数据
            if (message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue,message);
            }
            // 3.删除消息中心和确认集合中的消息
            memoryDataCenter.removeMessageWaitAsk(queueName,messageId);
            memoryDataCenter.removeMessage(messageId);
            System.out.println("[VirtualHost] bisicAck 成功，queueName = "+queueName+", messageId = "+messageId);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] bisicAck 失败，queueName = "+queueName+", messageId = "+messageId);
            e.printStackTrace();
            return false;
        }
    }
}

